package cn.doitedu.profile;

import cn.doitedu.eagle.beans.LogBean;
import cn.doitedu.demo.LogBeanFlat;
import cn.doitedu.profile.utils.LogBeanUtil;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Properties;

/**
 * @author 涛哥
 * @nick_name "deep as the sea"
 * @contact qq:657270652 wx:doit_edu
 * @site www.doitedu.cn
 * @date 2021-03-25
 * @desc 测试
 */
public class Main {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        Properties props = new Properties();

        props.setProperty("bootstrap.servers", "hdp01:9092,hdp02:9092,hdp03:9092");
        props.setProperty("auto.offset.reset", "earliest");
        FlinkKafkaConsumer<String> kafkaSource
                = new FlinkKafkaConsumer<String>("app_log", new SimpleStringSchema(), props);

        DataStreamSource<String> ds = env.addSource(kafkaSource);

        SingleOutputStreamOperator<LogBeanFlat> beanStream = ds.map(new MapFunction<String, LogBeanFlat>() {
            @Override
            public LogBeanFlat map(String value) throws Exception {
                LogBean logBean = JSON.parseObject(value, LogBean.class);
                LogBeanFlat flatBean = LogBeanUtil.flatBean(logBean);
                return flatBean;
            }
        });

        JdbcStatementBuilder<LogBeanFlat> builder = new JdbcStatementBuilder<LogBeanFlat>() {
            @Override
            public void accept(PreparedStatement st, LogBeanFlat bean) throws SQLException {
                st.setString(1, bean.getAccount());
                st.setString(2, bean.getAppId());
                st.setString(3, bean.getAppVersion());
                st.setString(4, bean.getCarrier());
                st.setString(5, bean.getDeviceId());
                st.setString(6, bean.getDeviceType());
                st.setString(7, bean.getEventId());
                st.setString(8, bean.getIp());
                st.setString(9, bean.getLatitude()+"");
                st.setString(10, bean.getLongitude()+"");
                st.setString(11, bean.getNetType());
                st.setString(12, bean.getOsName());
                st.setString(13, bean.getOsVersion());
                st.setString(14, bean.getReleaseChannel());
                st.setString(15, bean.getResolution());
                st.setString(16, bean.getSessionId());
                st.setString(17, bean.getTimeStamp());
                st.setString(18, bean.getAdCampain());
                st.setString(19, bean.getAdId());
                st.setString(20, bean.getAdLocation());
                st.setString(21, bean.getCouponId());
                st.setString(22, bean.getKeywords());
                st.setString(23, bean.getPageId());
                st.setString(24, bean.getProductId());
                st.setString(25, bean.getProductNumb());
                st.setString(26, bean.getRefType());
                st.setString(27, bean.getRefUrl());
                st.setString(28, bean.getShareMethod());
                st.setString(29, bean.getTitle());
                st.setString(30, bean.getUrl());
                st.setString(31, bean.getUtm_campain());
                st.setString(32, bean.getUtm_loctype());
                st.setString(33, bean.getUtm_source());
            }
        };


        String sql = "insert into xdb.event_detail values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
        JdbcExecutionOptions executionOptions = JdbcExecutionOptions.builder()
                .withBatchIntervalMs(100)
                .withBatchSize(5)
                .withMaxRetries(0)
                .build();

        JdbcConnectionOptions.JdbcConnectionOptionsBuilder connBuilder = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder();
        JdbcConnectionOptions connOptions = connBuilder
                .withUrl("jdbc:mysql://hdp01:9030/xdb")
                .withDriverName("com.mysql.jdbc.Driver")
                .withUsername("root")
                .withPassword("root").build();

        SinkFunction<LogBeanFlat> sink = JdbcSink.sink(sql,
                builder,
                connOptions
        );
        beanStream.print();

        beanStream.addSink(new MyJDBCSink());

        env.execute();
    }
}

class MyJDBCSink extends RichSinkFunction<LogBeanFlat> {
    //连接、定义预编译器
    Connection conn;
    PreparedStatement st;
    //PreparedStatement updateStmt;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        String sql = "insert into xdb.event_detail values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";

        conn = DriverManager.getConnection("jdbc:mysql://hdp01:9030/xdb","root","root");
        st = conn.prepareStatement(sql);

    }

    @Override
    public void invoke(LogBeanFlat bean, Context context) throws Exception {
        st.setString(1, bean.getAccount());
        st.setString(2, bean.getAppId());
        st.setString(3, bean.getAppVersion());
        st.setString(4, bean.getCarrier());
        st.setString(5, bean.getDeviceId());
        st.setString(6, bean.getDeviceType());
        st.setString(7, bean.getEventId());
        st.setString(8, bean.getIp());
        st.setString(9, bean.getLatitude()+"");
        st.setString(10, bean.getLongitude()+"");
        st.setString(11, bean.getNetType());
        st.setString(12, bean.getOsName());
        st.setString(13, bean.getOsVersion());
        st.setString(14, bean.getReleaseChannel());
        st.setString(15, bean.getResolution());
        st.setString(16, bean.getSessionId());
        st.setString(17, bean.getTimeStamp());
        st.setString(18, bean.getAdCampain());
        st.setString(19, bean.getAdId());
        st.setString(20, bean.getAdLocation());
        st.setString(21, bean.getCouponId());
        st.setString(22, bean.getKeywords());
        st.setString(23, bean.getPageId());
        st.setString(24, bean.getProductId());
        st.setString(25, bean.getProductNumb());
        st.setString(26, bean.getRefType());
        st.setString(27, bean.getRefUrl());
        st.setString(28, bean.getShareMethod());
        st.setString(29, bean.getTitle());
        st.setString(30, bean.getUrl());
        st.setString(31, bean.getUtm_campain());
        st.setString(32, bean.getUtm_loctype());
        st.setString(33, bean.getUtm_source());

        st.execute();

    }

    @Override
    public void close() throws Exception {
        conn.close();
        st.close();
        //updateStmt.close();
    }
}

